package com.atguigu.sparkcore.rdd.practices

import com.atguigu.sparkcore.util.MySparkContextUtil
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD

/**
  * 统计出每一个省份广告被点击次数的 TOP3
  * author 剧情再美终是戏
  * mail 13286520398@163.com
  * date 2020/1/7 15:38 
  * version 1.0
  **/
object ProvinceAdsTop3 {

  def main(args: Array[String]): Unit = {

    // 获取sparkContext
    val sc = MySparkContextUtil.get(args)

    // 创建rdd
    val rdd = sc.textFile("F:\\file\\尚硅谷相关资料\\003spark\\2019_08_30\\01_spark\\02_资料\\agent.log")

    // 转换rdd

    // 数据结构：时间戳，省份，城市，用户，广告，字段使用空格分割。
    //    1516609143867 6 7 64 16
    //    1516609143869 9 4 75 18
    //    1516609143869 1 7 87 12

    // 1、原始数-->((省份，广告)，1)
    // 2、对上一步进行reduceByKey --> ((省份，广告),count)
    // 3、对上一步进行map操作 -->  (省份 (广告,count))
    // 4、对上一步进行groupByKey --> (省份 List((广告,count))):取前3时，修改了value的值，key没有修改
    // 5、对上一步的count进行(mapvalues)排序，取top3
    // 6、对整体结果的省份进行排序

    val result = rdd.map {
      line =>
        val splits = line.split(" ")
        ((splits(1), splits(4)), 1)
    }.reduceByKey(_ + _)
      .map {
        case ((pro, ads), count) => (pro, (ads, count))
      }
      .groupByKey()
      .mapValues(x => x.toList.sortBy(_._2)(Ordering.Int.reverse).take(3))
      .sortBy(_._1.toInt)


    // 输出
    result.collect().foreach(println)

    // 关闭资源
    MySparkContextUtil.close(sc)
  }

}
